package com.shujia.core

import com.shujia.core.Demo10Join.Student
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

object Demo20Cache {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo20Cache")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val stuRDD: RDD[String] = sc
      .textFile("Spark/data/students.txt")

    val mapStuRDD: RDD[String] = stuRDD
      .map(line => {
        println("======读取了Students数据======")
        line
      })

    // 对多次使用的RDD进行cache 默认会将其缓存到Executor的内存中
    //    mapStuRDD.cache()
    // 如果想要使用其他的缓存策略 需要使用persist()方法
    //    mapStuRDD.persist(StorageLevel.MEMORY_ONLY)
    mapStuRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)

    // 统计班级人数
    val clazzRDD: RDD[(String, Int)] = mapStuRDD
      .map(line => (line.split(",")(4), 1))

    val clazzCntRDD: RDD[(String, Int)] = clazzRDD
      .reduceByKey(_ + _)

    // 打印
    clazzCntRDD.foreach(println)

    // 统计性别人数
    val genderRDD: RDD[(String, Int)] = mapStuRDD
      .map(line => (line.split(",")(3), 1))

    val genderCntRDD: RDD[(String, Int)] = genderRDD
      .reduceByKey(_ + _)

    // 打印
    genderCntRDD.foreach(println)

    // 使用完了之后记得释放缓存的RDD
    mapStuRDD.unpersist()

    while (true) {

    }

  }

}
